package com.kohburg.baseplate.xdelay.consumer;

import com.kohburg.baseplate.xdelay.config.DelayQueueConfig;
import com.kohburg.baseplate.xdelay.entity.XdelayMessage;
import com.kohburg.baseplate.xdelay.factory.RabbitTemplateFactory;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.IOException;

/**
 * 延迟队列的消费者
 *
 * @author : zenos hua
 * @date : 2020/6/17 18:07
 **/
@Component
public class XdelayConsumer {

    @Autowired
    private RabbitTemplateFactory rabbitTemplateFactory;

    @RabbitListener(
        bindings = @QueueBinding(
            value = @Queue(value = DelayQueueConfig.IMMEDIATE_QUEUE_XDELAY, durable = "true"),
            exchange = @Exchange(
                value = DelayQueueConfig.DELAYED_EXCHANGE_XDELAY,
                type = "x-delayed-message",
                durable = "true",
                ignoreDeclarationExceptions = "true",
                arguments = @Argument(name = "x-delayed-type", value = "direct")
            ),
            key = DelayQueueConfig.DELAY_ROUTING_KEY_XDELAY
        )
    )
    public void consume(XdelayMessage xdelayMsg, Channel channel) throws IOException {
        // 获取消息发送模板
        RabbitTemplate rabbitTemplate = rabbitTemplateFactory.getInstace();

        // 发送延迟消息
        rabbitTemplate.convertAndSend(xdelayMsg.getExchange(), xdelayMsg.getRoutingKey(), xdelayMsg.getMessage(), message -> {
            message.getMessageProperties().setDelay((int) xdelayMsg.getDelayTime());
            return message;
        });
    }
}
